package com.cnlive.logpare.play

import com.cnlive.logpare.comm.CommRun
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import com.cnlive.logpare.comm.CommNginxRun
import java.util.Date
import org.apache.spark.sql.Row
import scala.collection.mutable.HashSet
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.IntegerType
import org.apache.spark.sql.types.LongType
import org.apache.spark.sql.types.StringType
import scala.collection.mutable.HashMap
import com.cnlive.logpare.util.ParseLogUtil
import com.cnlive.logpare.util.Constant

object NewParsePlayLogToHive extends CommRun with CommNginxRun{
    val logSign="playBy"
    val defaultAppName="NewParsePlayLogToHive"
    def main(args: Array[String]): Unit = {
       val env=getEnv(args)
       val date=getDate(args)
       val files=getFilePath(env, logSign,date)
       val qqwryPath=getQqwryPath(env)//解析ip的文件地址
       val appname=getAppName(defaultAppName,date)
       val conf = new SparkConf().setAppName(appname)
       val installTime=tableNinxDateFormat.format(new Date) 
       if(env==1)conf.setMaster("local")
       val sc = new SparkContext(conf)
       val tabData= sc.textFile(files).mapPartitions(x=>{
         var result = List[Row]()
         val iPSeekerExt=getIpSeekerExt(env,qqwryPath)
         val sessionSet=new HashSet[String]
         while (x.hasNext) {
           val lineText=x.next()
           try{
               result= result.::(parLineNginxLogText(iPSeekerExt,lineText,sessionSet,parPalyLogUrlandtypeToMap,mkPlaylogRow,filter,installTime))
           }catch {
               case t: Throwable =>println("正行抛出异常"+lineText); t.printStackTrace() 
           }
         }
         result.iterator
       },true).filter(x => x.getInt(0)==1).repartition(1)
       //.foreach { println(_) }
       val spark=getSparkSql(env,conf)
       import spark.implicits._
       val palyLogtab=spark.createDataFrame(tabData, palyLogSchema)
       palyLogtab.registerTempTable("palyLogtab")
       val querySql="""
             select * from palyLogtab limit 10
                  """
       val installSql="""
             insert into TABLE  playofflog partition(year,moth,day) 
             select nginxtime,appid,spid,uid,ip,city,location,operator,plat,
             isnewuser,playtype,actitem,installtime,year,moth,day from  palyLogtab
                  """
       if(env==1){
          spark.sql(querySql).show()
       }else {
         spark.sql(installSql)
       }
      
    }
   val filter=(lineText:String)=>{
      if(lineText.indexOf(logSign) != -1){/*println(lineText);*/ true}  else false
   }
   val listPlayStr=List("liveplayByH5",
               "liveplayBySWF",
               "liveplayByAPP",
               "vodplayBySWF",
               "vodplayByH5",
               "vodplayByAPP"
               )
   val listPalyMap=Map("liveplayByH5"->(1,303),
                       "liveplayBySWF"->(1,301),
                       "liveplayByAPP"->(1,302),
                       "vodplayBySWF"->(2,301),
                       "vodplayByH5"->(2,303),
                       "vodplayByAPP"->(2,302)
    )
   //解析url中的数据到map中
   val  parPalyLogUrlandtypeToMap=(urlandtype:List[String],mapRequest:HashMap[String, String],sessionSet:HashSet[String])=>{
       val urlandper=urlandtype(0).split(" ")
       val d=urlandper match {
              case Array(_,_,z,_*) =>{
                  ParseLogUtil.parseUrlToMap(z, mapRequest)//解析url中的参数
                  listPlayStr.foreach(playType=>{
                    if(z.indexOf(playType) != -1){
                      mapRequest.put("playtype", listPalyMap.getOrElse(playType, (0,399))._1+"")
                      mapRequest.put("plat", listPalyMap.getOrElse(playType, (0,399))._2+"")
                      parIsNewUser(mapRequest,sessionSet)
                    }
                  })
              }
              case _=>mapRequest.put(Constant.errCodeKey, Constant.errCodeValue_001)
       }
   }
   //生成playLog的一行数据
   val mkPlaylogRow=(mapRequest:HashMap[String, String])=>{
            mapRequest.get(Constant.errCodeKey)  match {
                case Some(codeValue) =>if(codeValue.equals(Constant.errCodeValue_001)) Row(5)else Row(6)
                case _=>{
                    val appid=mapRequest.getOrElse("appid", "null")
                    val spidandappid=appid.split("_")
                    if(spidandappid.length!=2){
                        Row(4)
                    }else{
                       val nginxTime=tableNinxDateFormat.format(nginxDateFormat.parse(mapRequest.getOrElse("nginxtime", "19700101000000")))
                       val uid=mapRequest.getOrElse("uid", "0") 
                       val ip=mapRequest.getOrElse("ip", "127.0.0.1")
                       val spid=spidandappid(0)
                       val city=mapRequest.getOrElse("city", "unknown")//默认值
                       val location=mapRequest.getOrElse("location", "unknown")//默认值
                       val operator=mapRequest.getOrElse("operator", "99").toInt
                       val plat=mapRequest.getOrElse("plat", "399").toInt
                       val isnewuser=mapRequest.getOrElse("isnewuser", "0").toInt//默认值
                       val installtime=mapRequest.getOrElse("installtime", "19700101000000")//默认值
                       val playtype=mapRequest.getOrElse("playtype", "0").toInt//默认值
                       val actitem= playtype  match {
                          case 1 => mapRequest.getOrElse("channelid", "0")//直播
                          case 2 => mapRequest.getOrElse("vid", "0")//点播
                       }
                       Row(1,nginxTime.toLong,appid,spid,uid,ip,city,location,operator,plat,isnewuser,playtype,actitem,installtime.toLong,nginxTime.substring(0, 4),nginxTime.substring(4, 6),nginxTime.substring(6, 8))
                    }
                }
           }
    }
    //解析，看是不是一个新的用户
    def parIsNewUser(mapRequest:HashMap[String, String],sessionSet:HashSet[String]){
       if(mapRequest.contains("uid")&&mapRequest.contains("nginxtime")){
          val uid=mapRequest.getOrElse("uid", "")
          val tm=nginxDateFormat.parse(mapRequest.getOrElse("nginxtime", "19700101000000")).getTime
          val uidAndtm=uid.split("_")
          if(uidAndtm.length==2){
            if(uidAndtm(1).length()==13&&uidAndtm(1).matches("^[0-9]*$")){
              if((uidAndtm(1).toLong/1000*1000).equals(tm)){
                 if(sessionSet.contains(uid)){
                    mapRequest.put("isnewuser", "0")  
                 }else{
                    sessionSet.add(uid)
                    mapRequest.put("isnewuser", "1")  
                 }
               }else{
                  mapRequest.put("isnewuser", "0")
               }
            }else if(uidAndtm(0).length()==13&&uidAndtm(0).matches("^[0-9]*$")){
              if((uidAndtm(0).toLong/1000*1000).equals(tm)){
                 if(sessionSet.contains(uid)){
                    mapRequest.put("isnewuser", "0")   
                 }else{
                    sessionSet.add(uid)
                    mapRequest.put("isnewuser", "1")   
                 }
               }else{
                  mapRequest.put("isnewuser", "0")
               }
            }
          }else{
             mapRequest.put("isnewuser", "0")
          }
       }else{
         mapRequest.put("isnewuser", "0")
       }
    }
    //playLog表结构
    val palyLogSchema = StructType(Seq(
        StructField("datatype", IntegerType, nullable = true),
        StructField("nginxtime", LongType, nullable = true),
        StructField("appid", StringType, nullable = true),
        StructField("spid", StringType, nullable = true),
        StructField("uid", StringType, nullable = true),
        StructField("ip", StringType, nullable = true),
        StructField("city", StringType, nullable = true),
        StructField("location", StringType, nullable = false),
        StructField("operator", IntegerType, nullable = false),
        StructField("plat", IntegerType, nullable = false),
        StructField("isnewuser", IntegerType, nullable = false),
        StructField("playtype", IntegerType, nullable = false),
        StructField("actitem", StringType, nullable = false),
        StructField("installtime", LongType, nullable = true),
        StructField("year", StringType, nullable = true),
        StructField("moth", StringType, nullable = true),
        StructField("day", StringType, nullable = true)
      ))
}